package com.flink.hbase.elasticsearch2kafka;

import java.io.Serializable;
import java.util.List;
import java.util.Objects;

/**
 * Elasticsearch 分片分配信息
 */
public class ShardAssignment implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private final int taskIndex;
    private final List<ShardInfo> shards;
    
    public ShardAssignment(int taskIndex, List<ShardInfo> shards) {
        this.taskIndex = taskIndex;
        this.shards = shards;
    }
    
    public int getTaskIndex() {
        return taskIndex;
    }
    
    public List<ShardInfo> getShards() {
        return shards;
    }
    
    /**
     * 分片信息
     */
    public static class ShardInfo implements Serializable {
        private static final long serialVersionUID = 1L;
        
        private final String indexName;
        private final int shardId;
        private final String nodeId;
        private final String nodeName;
        private final boolean isPrimary;
        private final long documentCount;
        private final String state;
        
        public ShardInfo(String indexName, int shardId, String nodeId, String nodeName, 
                        boolean isPrimary, long documentCount, String state) {
            this.indexName = indexName;
            this.shardId = shardId;
            this.nodeId = nodeId;
            this.nodeName = nodeName;
            this.isPrimary = isPrimary;
            this.documentCount = documentCount;
            this.state = state;
        }
        
        public String getIndexName() {
            return indexName;
        }
        
        public int getShardId() {
            return shardId;
        }
        
        public String getNodeId() {
            return nodeId;
        }
        
        public String getNodeName() {
            return nodeName;
        }
        
        public boolean isPrimary() {
            return isPrimary;
        }
        
        public long getDocumentCount() {
            return documentCount;
        }
        
        public String getState() {
            return state;
        }
        
        /**
         * 获取分片标识符
         */
        public String getShardIdentifier() {
            return indexName + "_" + shardId + "_" + (isPrimary ? "primary" : "replica");
        }
        
        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            ShardInfo shardInfo = (ShardInfo) o;
            return shardId == shardInfo.shardId &&
                   isPrimary == shardInfo.isPrimary &&
                   Objects.equals(indexName, shardInfo.indexName);
        }
        
        @Override
        public int hashCode() {
            return Objects.hash(indexName, shardId, isPrimary);
        }
        
        @Override
        public String toString() {
            return "ShardInfo{" +
                    "indexName='" + indexName + '\'' +
                    ", shardId=" + shardId +
                    ", nodeId='" + nodeId + '\'' +
                    ", nodeName='" + nodeName + '\'' +
                    ", isPrimary=" + isPrimary +
                    ", documentCount=" + documentCount +
                    ", state='" + state + '\'' +
                    '}';
        }
    }
    
    @Override
    public String toString() {
        return "ShardAssignment{" +
                "taskIndex=" + taskIndex +
                ", shards=" + shards.size() +
                '}';
    }
} 